package edu.hcmus.sow.dao;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.springframework.stereotype.Repository;

@Repository
public class RepositoryDAOImpl extends AbstractCassandraDAO<String, Integer, byte[]> implements RepositoryDAO {
   private static final String KEYSPACE = "sow_repository";
   private static final int FRAGMENT_SIZE = 1000 * 1000; // 1MB

   public long saveFile(String columnFamily, String rowKey, InputStream stream) throws IOException {
      BufferedInputStream bis = new BufferedInputStream(stream);
      BufferedWriter writer = new BufferedWriter(KEYSPACE, columnFamily, rowKey, FRAGMENT_SIZE);
      byte[] fragment = new byte[FRAGMENT_SIZE];
      long size = 0;
      int len;

      while ((len = bis.read(fragment)) > 0) {
         writer.write(fragment, len);
         size += len;
      }
      writer.flush();
      return size;
   }

   public void readFile(String columnFamily, String rowKey, OutputStream stream) throws IOException {
      BufferedOutputStream bos = new BufferedOutputStream(stream);
      byte[] fragment;
      int index = 1;

      while ((fragment = fetchColumn(KEYSPACE, columnFamily, rowKey, index)) != null) {
         bos.write(fragment);
         index++;
      }
      bos.flush();
   }

   public void deleteFile(String columnFamily, String rowKey) {
      delete(KEYSPACE, columnFamily, rowKey);
   }

   public void createFileStorage(String columnFamily) {
      createColumnFamily(KEYSPACE, columnFamily);
   }

   public void dropFileStorage(String columnFamily) {
      dropColumnFamily(KEYSPACE, columnFamily);
   }

   private class BufferedWriter {
      private String keyspace;
      private String columnFamily;
      private String rowKey;
      private byte[] buf;
      private int count = 0, index = 1;

      public BufferedWriter(String keyspace, String columnFamily, String rowKey, int size) {
         this.keyspace = keyspace;
         this.columnFamily = columnFamily;
         this.rowKey = rowKey;
         this.buf = new byte[size];
      }

      public void write(byte[] bytes, int len) {
         // System.out.println("len: " + len);
         int bufRemainder = buf.length - count;

         if (len < bufRemainder) {
            System.arraycopy(bytes, 0, buf, count, len);
            count += len;
            return;
         }

         System.arraycopy(bytes, 0, buf, count, bufRemainder);
         count += bufRemainder;
         flush();

         if (len > bufRemainder) {
            System.arraycopy(bytes, bufRemainder, buf, count, len - bufRemainder);
            count += len - bufRemainder;
         }
         // System.out.println("count: " + count);
      }

      public void flush() {
         // System.out.println("flush: " + count);
         // System.out.println("flush " + rowKey + ": " + index + ": " + count);
         if (count > 0) {
            byte[] bytes = buf;
            if (count < buf.length) {
               bytes = new byte[count];
               System.arraycopy(buf, 0, bytes, 0, count);
            }

            persist(keyspace, columnFamily, rowKey, index++, bytes);
            count = 0;
         }
      }
   }
}